package ads;

import bean.Profiletags;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.DorisUtil;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;



//                {
//                    "weight_records": [
//                    {
//                        "source_label": "智能体脂秤",
//                            "source_weight": 1.0,
//                            "weight": 59.0,
//                            "source": "health_device",
//                            "timestamp": 1749449752680
//                    },
//                    {
//                        "source_label": "健康商品订单",
//                            "source_weight": 0.8,
//                            "weight": 58.7,
//                            "source": "order_info",
//                            "timestamp": 1748326552680
//                    },
//                    {
//                        "source_label": "会员资料填写",
//                            "source_weight": 0.6,
//                            "weight": 58.4,
//                            "source": "member_profile",
//                            "timestamp": 1749449752680
//                    },
//                    {
//                        "source_label": "活动表单提交",
//                            "source_weight": 0.4,
//                            "weight": 59.7,
//                            "source": "campaign_form",
//                            "timestamp": 1747635352680
//                    }
//  ],
//                    "cart_count": 42,
//                        "price_sensitivity": 0.18,
//                        "gender": "家庭用户",
//                        "birth_month": "",
//                        "weight": 38.3,
//                        "purchase_count": 94,
//                        "age_label": "50岁以上",
//                        "time_behavior": {
//                    "afternoon": 7,
//                            "night": 14,
//                            "recent_activity_level": "high",
//                            "morning": 5
//                },
//                    "process_time": "2025-06-11T14:16:36.282",
//                        "birth_year": "",
//                        "birth_day": "",
//                        "search_terms": "笔记本电脑 零食 育儿经验",
//                        "height_records": [
//                    {
//                        "source_label": "智能设备",
//                            "source_weight": 1.0,
//                            "source": "smart_device",
//                            "height": 161.4,
//                            "timestamp": 1748844952680
//                    },
//                    {
//                        "source_label": "实名认证",
//                            "source_weight": 0.9,
//                            "source": "real_name_auth",
//                            "height": 162.0,
//                            "timestamp": 1748412952680
//                    },
//                    {
//                        "source_label": "近30天订单",
//                            "source_weight": 0.85,
//                            "source": "recent_order",
//                            "height": 162.3,
//                            "timestamp": 1748240152680
//                    },
//                    {
//                        "source_label": "历史订单",
//                            "source_weight": 0.7,
//                            "source": "history_order",
//                            "height": 90.5,
//                            "timestamp": 1749190552680
//                    },
//                    {
//                        "source_label": "客服工单",
//                            "source_weight": 0.6,
//                            "source": "customer_service",
//                            "height": 259.0,
//                            "timestamp": 1747376152680
//                    }
//  ],
//                    "brand_pref": "快时尚",
//                        "device_info": "iPhone",
//                        "constellation": "未知",
//                        "user_id": 190400,
//                        "category_pref": "食品生鲜",
//                        "browse_count": 376,
//                        "social_interaction": 49,
//                        "height": 161.1,
//                        "timestamp": 1749622552680
//                }
public class dwd_user_profile {
    public static void main(String[] args)throws Exception {
        //1.创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 启用 Checkpointing（例如每5秒一次）
//        env.enableCheckpointing(5000);
        //2.从kafka中读取数据
        DataStream<String> dataStream = KafkaUtil.consumerKafka(env, "user_profile_processed_tags");
        //3.打印数据
//        dataStream.print();
        //4.处理数据
        DataStream<String> handle = handle(dataStream);
//        handle.print();
        //5.保存到doris
        DorisUtil.saveToDoris(handle,"jtp_label","user_profile_tags");


        env.execute("JtpLogEtIJob");

    }

    private static DataStream<String> handle(DataStream<String> dataStream) {
        SingleOutputStreamOperator<Profiletags> cartCount = dataStream.map(new MapFunction<String, Profiletags>() {
            @Override
            public Profiletags map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String userId = jsonObject.getString("user_id");
                String age_label = jsonObject.getString("age_label");
                String gender = jsonObject.getString("gender");

                String birth_year =null;
                //  处理出生年代
                if (!jsonObject.getString("birth_year").isEmpty()){
                    birth_year=jsonObject.getString("birth_year").substring(2)+"年代";
                }else {
                    birth_year="未知";
                }
                String constellation = jsonObject.getString("constellation");
                Profiletags profiletags = new Profiletags();
                //  处理体重
                Double weight=weightrecords(jsonObject);
                //  处理身高
                Double height=heightrecords(jsonObject);
                //  处理年龄
//                System.out.println(weight);
                profiletags.setWeight(weight);
                profiletags.setTs(new Date().getTime());
                profiletags.setUser_id(userId);
                profiletags.setAge_range(age_label);
                profiletags.setGender(gender);
                profiletags.setHeight(height);
                profiletags.setGeneration(birth_year);
                profiletags.setZodiac_sign(constellation);
                return profiletags;
            }

            private Double heightrecords(JSONObject jsonObject) {

                double  sum=0.0;
                double  count=0.0;
                JSONArray heightRecords = jsonObject.getJSONArray("height_records");
                for (Object heightRecord : heightRecords) {
                    JSONObject jsonObject1 = JSON.parseObject(heightRecord.toString());
                    if (jsonObject1.getString("source_label").equals("智能设备")){
                        sum+=jsonObject1.getDouble("height")*jsonObject1.getDouble("source_weight");
                        count+=jsonObject1.getDouble("source_weight");
                    }else if (jsonObject1.getString("source_label").equals("实名认证")){
                        sum+=jsonObject1.getDouble("height")*jsonObject1.getDouble("source_weight");
                        count+=jsonObject1.getDouble("source_weight");

                    }else if (jsonObject1.getString("source_label").equals("近30天订单")){
                        sum+=jsonObject1.getDouble("height")*jsonObject1.getDouble("source_weight");
                        count+=jsonObject1.getDouble("source_weight");
                    }else {
                        sum+=jsonObject1.getDouble("height")*jsonObject1.getDouble("source_weight");
                        count+=jsonObject1.getDouble("source_weight");
                    }

                }


                return sum/count;
            }

            private Double weightrecords(JSONObject jsonObject) {
                double  sum=0.0;
                double  count=0.0;
                JSONArray weightRecords = jsonObject.getJSONArray("weight_records");
                for (Object weightRecord : weightRecords) {
                    JSONObject jsonObject1 = JSON.parseObject(weightRecord.toString());
                    if (jsonObject1.getString("source_label").equals("智能体脂秤")){
                        sum+=jsonObject1.getDouble("weight")*jsonObject1.getDouble("source_weight");
                        count+=jsonObject1.getDouble("source_weight");
                    }else if (jsonObject1.getString("source_label").equals("健康商品订单")){
                        sum+=jsonObject1.getDouble("weight")*jsonObject1.getDouble("source_weight");
                        count+=jsonObject1.getDouble("source_weight");
                    }else if (jsonObject1.getString("source_label").equals("会员资料填写")){
                        sum+=jsonObject1.getDouble("weight")*jsonObject1.getDouble("source_weight");
                        count+=jsonObject1.getDouble("source_weight");
                    }else {
                        sum+=jsonObject1.getDouble("weight")*jsonObject1.getDouble("source_weight");
                        count+=jsonObject1.getDouble("source_weight");
                    }
                }
                return sum/count;
            }


        });

//        cartCount.print();


        //3.设置水位线
        DataStream<Profiletags> timeStream = cartCount.assignTimestampsAndWatermarks(WatermarkStrategy
                // 设置延迟时间
                .<Profiletags>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner(
                        new SerializableTimestampAssigner<Profiletags>() {
                            @Override
                            public long extractTimestamp(Profiletags element, long recordTimestamp) {
                                if (element == null) {
                                    // 可以选择忽略该事件，或者使用当前系统时间作为替代
                                    return System.currentTimeMillis(); // 或者 -1 表示无效
                                }
                                return element.getTs();
                            }
                        })
        );
//        4.按照维度进行分组
        KeyedStream<Profiletags, String> keyedStream = timeStream.keyBy(bean -> bean.getUser_id());
        //5.开窗
//        keyedStream.print();
        WindowedStream<Profiletags, String, TimeWindow> windowStream = keyedStream.window(
                TumblingEventTimeWindows.of(Time.minutes(1))
        );

        SingleOutputStreamOperator<String> apply = windowStream.apply(new WindowFunction<Profiletags, String, String, TimeWindow>() {
            private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

            @Override
            public void apply(String key, TimeWindow window, Iterable<Profiletags> input, Collector<String> out) throws Exception {
                Profiletags profiletags = input.iterator().next();
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("user_id", profiletags.getUser_id());
                jsonObject.put("age_range", profiletags.getAge_range());
                jsonObject.put("generation", profiletags.getGeneration());
                jsonObject.put("gender", profiletags.getGender());
                jsonObject.put("weight", profiletags.getWeight());
                jsonObject.put("height", profiletags.getHeight());
                jsonObject.put("zodiac_sign", profiletags.getZodiac_sign());
                jsonObject.put("ts", format.format(profiletags.getTs()));

                String output = String.format(
                        "%s,%s,%s,%s,%s,%s,%s",
                        profiletags.getUser_id(),
                        profiletags.getAge_range(),
                        profiletags.getGender(),
                        profiletags.getWeight(),
                        profiletags.getHeight(),
                        profiletags.getGeneration(),
                        profiletags.getZodiac_sign(),
                        format.format(profiletags.getTs())  // 格式化时间戳
                );

                // 发送结果
                out.collect(jsonObject.toString());
            }


        });
//        apply.print();
        return apply;

    }
}
